package com.atguigu.flink.wordcount;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import static org.apache.flink.api.common.typeinfo.Types.*;

/**

 *
 */
public class Demo8_LamdaTuple
{
    public static void main(String[] args) throws Exception {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        //全局，所有的算子，都使用1个并行度
        env.setParallelism(1);

        DataStreamSource<String> source = env.socketTextStream("hadoop103",8888);


        SingleOutputStreamOperator<Tuple2<String, Integer>> ds1 = source
            .flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (line, out) -> {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(new Tuple2<>(word, 1));
                }
            })
            /*
                returns(Class<T> typeClass): 仅仅适用于 T类型中没有泛型的情况。
                如果T是泛型类型，例如Tuple,必须使用:
                    returns(TypeHint<T> typeHint)
                        或
                    returns(TypeInformation<T> typeInfo)
                            Types是一个工具类，可以快速返回常见类型的TypeInformation对象。
                            所有的静态内容(方法，属性)都可以静态导入，导入后，在这个类中使用静态内容，无需类名。

             */
            //.returns(new TypeHint<Tuple2<String, Integer>>(){});
            .returns(TUPLE(STRING,INT));

        ds1
            .keyBy((KeySelector<Tuple2<String, Integer>, String>) value -> value.f0)
            .sum(1)
            .print();

        env.execute();

    }
}
